package com.zidongxiangxi.rabbit.producer.interceptor;

import com.zidongxiangxi.rabbit.core.constant.enums.MessageTypeEnum;
import com.zidongxiangxi.rabbit.core.exception.ReliableRabbitException;
import com.zidongxiangxi.rabbit.core.util.RabbitUtils;
import com.zidongxiangxi.rabbit.producer.constant.ProducerConstants;
import com.zidongxiangxi.rabbit.producer.entity.Producer;
import com.zidongxiangxi.rabbit.producer.entity.RabbitMessageInfo;
import com.zidongxiangxi.rabbit.producer.exception.NotSupportOperationException;
import com.zidongxiangxi.rabbit.producer.manager.ProducerManager;
import com.zidongxiangxi.rabbit.producer.support.RabbitMessagePostProcessor;
import com.zidongxiangxi.rabbit.producer.transaction.RabbitTransactionSynchronization;
import com.zidongxiangxi.rabbit.producer.transaction.listener.TransactionListener;
import com.zidongxiangxi.rabbit.producer.transaction.support.RabbitMessageTransactionContext;
import com.zidongxiangxi.rabbit.producer.util.ProducerUtils;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;

/**
 * RabbitTemplate方法拦截器
 *
 * @author chenxudong
 * @date 2020/11/07
 */
public class RabbitTemplateMethodInterceptor implements MethodInterceptor {
	private static final String SEND_METHOD_NAME = "send";
	private static final String CORRELATION_CONVERT_AND_SEND = "correlationConvertAndSend";
	private static final String CONVERT_AND_SEND_METHOD_NAME = "convertAndSend";
	private static final String SEND_AND_RECEIVE_METHOD_NAME = "sendAndReceive";
	private static final String CONVERT_SEND_AND_RECEIVE_METHOD_NAME = "convertSendAndReceive";

	private static final Set<String> INTERCEPTOR_METHOD_SET;

	static {
		INTERCEPTOR_METHOD_SET = new HashSet<>();
		INTERCEPTOR_METHOD_SET.add(SEND_METHOD_NAME);
		INTERCEPTOR_METHOD_SET.add(CORRELATION_CONVERT_AND_SEND);
		INTERCEPTOR_METHOD_SET.add(CONVERT_AND_SEND_METHOD_NAME);
		INTERCEPTOR_METHOD_SET.add(SEND_AND_RECEIVE_METHOD_NAME);
		INTERCEPTOR_METHOD_SET.add(CONVERT_SEND_AND_RECEIVE_METHOD_NAME);
	}

	private final String application;
	private final ProducerManager producerManager;
	private final TransactionSynchronization synchronization;
	private final RabbitMessagePostProcessor defaultMessagePostProcessor;

	public RabbitTemplateMethodInterceptor(
			String application,
			ProducerManager producerManager,
			List<TransactionListener> transactionListener
	) {
		this.application = application;
		this.producerManager = producerManager;
		this.synchronization = new RabbitTransactionSynchronization(transactionListener);
		this.defaultMessagePostProcessor = new RabbitMessagePostProcessor(application);
	}

	@Override
	public Object invoke(MethodInvocation invocation) throws Throwable {
		Method method = invocation.getMethod();
		String methodName = method.getName();
		if (!INTERCEPTOR_METHOD_SET.contains(methodName)) {
			return invocation.proceed();
		}
		Object target = invocation.getThis();
		if (!(target instanceof RabbitTemplate)) {
			return invocation.proceed();
		}
		boolean isPublicMethod = Modifier.PUBLIC == method.getModifiers();
		if (!isPublicMethod) {
			return invocation.proceed();
		}

		RabbitTemplate rabbitTemplate = (RabbitTemplate)target;
		Object[] args = invocation.getArguments();
		Class<?>[] argTypes = method.getParameterTypes();
		Optional<CorrelationData> correlationDataOptional =
				ProducerUtils.tryGetFirstArg(args, argTypes, CorrelationData.class);
		if (Objects.nonNull(correlationDataOptional) && correlationDataOptional.isPresent()) {
			throw new ReliableRabbitException("可靠MQ不支持指定CorrelationData");
		}

		Producer producer = new Producer();
		producer.setType(MessageTypeEnum.IMMEDIATE.getValue());
		producer.setApplication(application);
		producer.setHost(rabbitTemplate.getConnectionFactory().getHost());
		producer.setPort(rabbitTemplate.getConnectionFactory().getPort());
		producer.setVirtualHost(rabbitTemplate.getConnectionFactory().getVirtualHost());
		producer.setMaxRetryTimes(ProducerConstants.MAX_RETRY_TIMES);

		RabbitMessageInfo rabbitMessage;
		if (Objects.equals(methodName, SEND_METHOD_NAME)) {
			// 获取参数
			String exchange = null, routingKey = null;
			Optional<String> firstStringOptional = ProducerUtils.tryGetFirstArg(args, argTypes, String.class);
			Optional<String> sendStringOptional = ProducerUtils.tryGetSecondArg(args, argTypes, String.class);
			if (Objects.nonNull(firstStringOptional)) {
				if (Objects.nonNull(sendStringOptional)) {
					exchange = firstStringOptional.orElse(null);
					routingKey = sendStringOptional.orElse(null);
				} else {
					routingKey = firstStringOptional.orElse(null);
				}
			}
			Optional<Message> messageOptional = ProducerUtils.tryGetFirstArg(args, argTypes, Message.class);
			if (Objects.isNull(messageOptional) || !messageOptional.isPresent()) {
				return invocation.proceed();
			}
			// 判断是否顺序消息，修改producer类型
			Message message = messageOptional.get();
			if (RabbitUtils.isSequenceMessage(message)) {
				producer.setType(MessageTypeEnum.SEQUENCE.getValue());
			}
			// MessagePostProcessor加工Message
			Message realMessage = defaultMessagePostProcessor.postProcessMessage(message);
			CorrelationData correlationData = ProducerUtils.generateCorrelationData(application,
					realMessage.getMessageProperties().getMessageId());
			// 为producer设置缺少的属性
			producer.setMessageId(realMessage.getMessageProperties().getMessageId());
			producer.setExchange(exchange);
			producer.setRoutingKey(routingKey);
			producer.setMessage(convertMessageToBytes(message));
			// 组装RabbitMessageInfo
			rabbitMessage = new RabbitMessageInfo(producer, rabbitTemplate, exchange, routingKey, realMessage, correlationData);
		} else if (Objects.equals(methodName, CONVERT_AND_SEND_METHOD_NAME)) {
			// 获取参数
			Optional<Object> messageOptional = ProducerUtils.tryGetFirstArg(args, argTypes, Object.class);
			if (Objects.isNull(messageOptional) || Objects.isNull(messageOptional.orElse(null))) {
				return invocation.proceed();
			}
			Object message = messageOptional.get();
			MessagePostProcessor messagePostProcessor = null;
			String exchange = null, routingKey = null;
			Optional<MessagePostProcessor> messagePostProcessorOptional =
					ProducerUtils.tryGetFirstArg(args, argTypes, MessagePostProcessor.class);
			if (Objects.nonNull(messagePostProcessorOptional)) {
				messagePostProcessor = messagePostProcessorOptional.orElse(null);
			}
			Optional<String> firstStringOptional = ProducerUtils.tryGetFirstArg(args, argTypes, String.class);
			Optional<String> sendStringOptional = ProducerUtils.tryGetSecondArg(args, argTypes, String.class);
			if (Objects.nonNull(firstStringOptional)) {
				if (Objects.nonNull(sendStringOptional)) {
					exchange = firstStringOptional.orElse(null);
					routingKey = sendStringOptional.orElse(null);
				} else {
					routingKey = firstStringOptional.orElse(null);
				}
			}
			// MessagePostProcessor加工Message
			Message realMessage = rabbitTemplate.getMessageConverter().toMessage(message, new MessageProperties());
			if (Objects.nonNull(messagePostProcessor)) {
				messagePostProcessor.postProcessMessage(realMessage, null);
			}
			realMessage = defaultMessagePostProcessor.postProcessMessage(realMessage);
			CorrelationData correlationData = ProducerUtils.generateCorrelationData(application,
					realMessage.getMessageProperties().getMessageId());
			// 为producer设置缺少的属性
			producer.setMessageId(realMessage.getMessageProperties().getMessageId());
			producer.setExchange(exchange);
			producer.setRoutingKey(routingKey);
			producer.setMessage(convertMessageToBytes(realMessage));
			// 组装RabbitMessageInfo
			rabbitMessage = new RabbitMessageInfo(producer, rabbitTemplate, exchange, routingKey, realMessage, correlationData);
		} else {
			throw new NotSupportOperationException(methodName);
		}

		// 降级兜底机制
		// 1、优先使用事务同步器，将mq消息跟随业务数据一起落库
		// 2、如果不支持事务同步器，则直接单独保存mq消息
		// 3、如果不支持保存mq消息到数据库，则退化为普通的mq发送
		if (canUseTransactionSynchronization()) {
			registerTransactionSynchronization();
			RabbitMessageTransactionContext.stashMessage(rabbitMessage);
		} else {
			producerManager.saveProducer(rabbitMessage.getProducer());
			rabbitTemplate.send(rabbitMessage.getExchange(), rabbitMessage.getRoutingKey(),
					rabbitMessage.getMessage(), rabbitMessage.getCorrelationData());
			return null;
		}
		return null;
	}

	private byte[] convertMessageToBytes(Message message) throws IOException {
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		ObjectOutputStream oout = new ObjectOutputStream(baos);
		oout.writeObject(message);
		oout.close();
		return baos.toByteArray();
	}

	private boolean canUseTransactionSynchronization() {
		return TransactionSynchronizationManager.isSynchronizationActive()
				&& TransactionSynchronizationManager.isActualTransactionActive();
	}

	private void registerTransactionSynchronization() {
		TransactionSynchronizationManager.registerSynchronization(synchronization);
	}
}
